package edu.nepu.flink.api.window;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Date 2024/2/29 21:49
 * @Created by chenshuaijun
 */
public class TimeTumblingWindow {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
            }
        });

        SingleOutputStreamOperator<WaterSensor> process = source.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5L)))
//                .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
//                    @Override
//                    public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
//                        TimeWindow window = context.window();
//                        String windowStart = DateFormatUtils.format(window.getStart(), "yyyy-MM-dd HH:mm:ss");
//                        String windowEnd = DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
//                        out.collect(windowStart + "-->" + windowEnd + "  数据条数：" + elements.spliterator().estimateSize() + "  其中的元素" + elements.toString());
//                    }
//                });

        .reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                System.out.println(value1 + "--->" + value2);
                return new WaterSensor(value1.id,value1.ts,value1.vc + value2.vc);
            }
        });
        process.print();

        env.execute();
    }
}
